#include "raft.h"

#include "applyMsg.h"
#include "config.h"
#include "ratfRPC.pb.h"

#include <algorithm>
#include <chrono>
#include <glog/logging.h>
#include <memory>
#include <mutex>
#include <thread>
#include <unistd.h>
#include <vector>

namespace raftStoreDataBase
{
    void Raft::AppendEntries( google::protobuf::RpcController *controller,
                              const raftRpcProtoc::AppendEntriesArgs *request,
                              raftRpcProtoc::AppendEntriesReply *response,
                              google::protobuf::Closure *done )
    {
        AppendEntriesImp( request, response );
        done->Run(); // done就是绑定发送RPCrespnse的回调函数
    }

    /**
     *
     * @param logIndex 这个函数的作用是将logIndex映射成快照上传后的Index，因为快照上传后就会把快照上传的index给清除，但是logIndex是不会变得
     * 因此，我们需要将logIndex映射到正确的index处
     * @return
     */
    int Raft::getSlicesIndexFromLogIndex( int logIndex )
    {
        myAssert( logIndex > m_lastSnapShotIndex,
                  format( "[func-getSlicesIndexFromLogIndex-rf{%d}]  index{%d} <= rf.lastSnapshotIncludeIndex{%d}",
                          m_id,
                          logIndex,
                          m_lastSnapShotIndex ) );
        int lastLogIndex = getLastLogIndex();
        myAssert( logIndex <= lastLogIndex,
                  format( "[func-getSlicesIndexFromLogIndex-rf{%d}]  logIndex{%d} > lastLogIndex{%d}", m_id, logIndex, lastLogIndex ) );
        int slicesIndex = logIndex - m_lastSnapShotIndex - 1; // 减1是因为重新映射的下标是从0开始的
        return slicesIndex;
    }

    void Raft::AppendEntriesImp( const raftRpcProtoc::AppendEntriesArgs *request, raftRpcProtoc::AppendEntriesReply *response )
    {
        std::lock_guard< std::mutex > locker( m_mutex );
        response->set_appstate( AppNormal ); // 这是将appenstate状态设置为正常，这个response会发送给领导者让你判断follower的状态

        // 从leader发送的任期和自己的任期进行比较
        if (request->term() < m_currentTerm)
        {
            // 领导者的任期小于我的任期，我就要设置AppendEntries状态为false，并且将自己的任期返回给leader；
            response->set_success( false );
            response->set_term( m_currentTerm );
            response->set_updatenextindex( -100 ); // 这就是让领导者知道自己已经不是领导者了；
            std::cerr << "[func_AppendEntries-raftFollower" << m_id << "] 拒绝了AppendEntries，因为" << request->leaderid() << "的term("
                << request->term() << ") < raftFollower" << m_id << "的term(" << m_currentTerm << std::endl;
            LOG( INFO ) << "[func_AppendEntries-raftFollower" << m_id << "] 拒绝了AppendEntries，因为" << request->leaderid() << "的term("
                << request->term() << ") < raftFollower" << m_id << "的term(" << m_currentTerm;
            return;
        }

        DEFER { persist(); }; // 这是延迟更新的
        // 如果leader的term大于我的term，那么我自己的状态就是follower
        if (request->term() > m_currentTerm)
        {
            m_role = Follower;
            m_currentTerm = request->term();
            m_vectorFor = -1; // 这里设置成-1有意义，如果突然宕机然后上线理论上是可以投票的
        }

        myAssert( request->term() == m_currentTerm, format( "assert {args.Term == rf.currentTerm} fail" ) );

        // 这行代码确保无论当前节点的角色是什么，当收到有效的请求时，都会将其转换为
        // Follower。这是为了应对可能的网络分区或其他不一致的情况，确保节点总是回到一个安全的状态。
        m_role = Follower;

        // 设置选举超时,每次收到AppendEntries消息是都会重置选举超时计时器；表明节点已经在当前任期内接收到了有效的领导者消息，不需要发起新的选举。
        m_lastResetElectionTime = now();

        // 执行到这里就需要开始日志比较了，如果prelogIndex不在我的logIndex之内，则说明上一次的日志更新我没有参与，我都不包含上次的更新的日志条目；
        if (request->prelogindex() > getLastLogIndex())
        {
            response->set_success( false );
            response->set_term( m_currentTerm );
            response->set_updatenextindex( getLastLogIndex() + 1 );
            return;
        } // 如果你发送过来的logIndex都小于我的快照里面的最后一个logIndex，说明我需要的是，你发送的这个数据我都已经存在了，我需要快照之后的数据，就向leader申请我需要快照之后的log；
        else if (request->prelogindex() < m_lastSnapShotIndex)
        {
            response->set_success( false );
            response->set_term( m_currentTerm );
            response->set_updatenextindex( m_lastSnapShotIndex + 1 );
        }

        // 这里就是匹配你发送的prelogindex()是否存在我的本地中，如果存在的话就需要比较我存在的这个日志条目的term是否和你发送过来的一样
        // 即prelogindex()需要存在于本地，并且本地的prelogindex()的日志条目的term==prelogterm()；
        if (mathcLog( request->prelogindex(), request->prelogterm() ))
        {
            // 如果是匹配，则开始进行进行日志的更新，必须要一个一个的更新，不能直接添加，因为可能会包含我已经有的，所以必须要大于我的我才添加，等于不行，因为我已经存在了这个日志

            // 循环遍历logEntries
            for (auto &log : request->entries())
            {
                // 如果这个index不存在于我的log里面则直接添加，否则需要判断我保存的日志里面的index和term是否和你发送过来的一致，不是的话就添加
                if (log.logindex() > getLastLogIndex()) { m_logs.emplace_back( log ); }
                else
                {
                    // 如果不是，则需要看是否和我本地的日志里面的一摸一样，包括index和term，如果是相同的，但是里面的command是不一样的，则出现了异常
                    if (m_logs[ getSlicesIndexFromLogIndex( log.logindex() ) ].logterm() == log.logterm()
                        && m_logs[ getSlicesIndexFromLogIndex( log.logindex() ) ].command() != log.command())
                    {
                        myAssert( false,
                                  format( "[func-AppendEntries-rf{%d}] 两节点logIndex{%d}和term{%d}相同，但是其command{%d:%d}   "
                                          " {%d:%d}却不同！！\n",
                                          m_id,
                                          log.logindex(),
                                          log.logterm(),
                                          m_id,
                                          m_logs[ getSlicesIndexFromLogIndex( log.logindex() ) ].command(),
                                          request->leaderid(),
                                          log.command() ) );
                    }

                    // 如果里面的term不匹配了，则需要更新这个不匹配的
                    if (m_logs[ getSlicesIndexFromLogIndex( log.logindex() ) ].logterm() != log.logterm())
                    {
                        m_logs[ getSlicesIndexFromLogIndex( log.logindex() ) ] = log;
                    }
                }
            }

            // 当出现网络波动或者领导者故障恢复时，可能会发送之前已经发送了的AppendEntries请求，但是follower已经处理了之前发送的AppendEntries请求，
            // 这个请求在发送后被延迟到达。与此同时，追随者可能已经从领导者那里接收并提交了更多的日志条目。
            // 就是添加的条目一定会大于等于request里面的index+logEnrty的大小的，因为已经是匹配的，如果如果里面包含了我已经存在的index，那么添加完毕之后的最后一个index也是大于等于他们相加的
            // 就是可能会收到过期的日志，因此就是我必须大于等于
            myAssert( getLastLogIndex() >= request->prelogindex() + request->entries_size(),
                      format( "[func-AppendEntries1-rf{%d}]rf.getLastLogIndex(){%d} != args.PrevLogIndex{%d}+len(args.Entries){%d}",
                              m_id,
                              getLastLogIndex(),
                              request->prelogindex(),
                              request->entries_size() ) );

            // 接下来就需要更新提交的索引了
            if (request->leadercommit() > m_commitIndex)
            {
                // 两种情况：1、leadercommit是比我getLastLogIndex()小的，说明leader将这部分的日志条目已经提交了一部分，因此我们需要和leader一致；
                //          2、leadercommit >
                //          getLastLogIndex()，说明leader已经提交了这部分的日志，我们也需要把这部分的日志全部提交上去呀；
                // 即如果leader发送的request->leadercommit()大于getLastLogIndex()，则m_commitIndex等于当前的getLastLogIndex()的；
                // 如果小于的话，则就跟着leader一样提交；
                m_commitIndex = std::min( request->leadercommit(), getLastLogIndex() );
            }

            // 最后一个getLastLogIndex()一定会大于等于m_commitIndex
            myAssert(
                getLastLogIndex() >= m_commitIndex,
                format(
                    "[func-AppendEntries1-rf{%d}]  rf.getLastLogIndex{%d} < rf.commitIndex{%d}",
                    m_id,
                    getLastLogIndex(),
                    m_commitIndex ) );

            response->set_success( true );
            response->set_term( m_currentTerm );
            return;
        }
        else
        {
            // 这里是leader发送过来的request里面的preindex和preterm和我这里面存储的最后一个的index和term不匹配
            // 你需要重新发送一个
            // 如果从你发送过来的prelogindex()，然后从我的日志条目里面获取的log的term不等于你发送的prelogterm，然后我就设置这个
            response->set_updatenextindex( request->prelogindex() ); // 你重新发送一个过来，这种情况通常是leader崩溃了之后的

            // 然后就需要继续判断我是否从prelogindex()往前找，是否还有不匹配的情况
            for (int index = request->prelogindex(); index >= m_lastSnapShotIndex; --index)
            {
                if (getLogTermFromLogIndex( index ) != getLogTermFromLogIndex( request->prelogindex() ))
                {
                    response->set_updatenextindex( index + 1 );
                    break;
                }
            }

            response->set_success( false );
            response->set_term( m_currentTerm );
            return;
        }
    }

    /**
     * @brief 这是当对端收到了远程过程调用请求时，更新服务和方法描述符执行的请求投票的rpc调用
     *
     * @param controller
     * @param request
     * @param response
     * @param done
     */
    void Raft::RequestVote( google::protobuf::RpcController *controller,
                            const raftRpcProtoc::RequestVoteArgs *request,
                            raftRpcProtoc::RequestVoteReply *response,
                            google::protobuf::Closure *done )
    {
        RequestVoteImp( request, response );
        done->Run();
    }

    /**
     * @brief 这里是对端执行candidate发送过来的请求投票请求，并将投票结果填写到response中，然后通过回调函数发送回对端；
     *
     * @param request 这是candidate发送过来的自己的参数
     * @param response 这是投票者对投票的响应
     */
    void Raft::RequestVoteImp( const raftRpcProtoc::RequestVoteArgs *request, raftRpcProtoc::RequestVoteReply *response )
    {
        // 加锁
        std::lock_guard< std::mutex > lock( m_mutex );

        // 持久化存储
        DEFER { persist(); };

        // 我需要判断发送过来的RequestVoteArgs里面的任期和我的term相比较的情况

        // 小于的情况，出现网络分区时
        if (request->term() < m_currentTerm)
        {
            response->set_term( m_currentTerm );
            response->set_votegranted( false );
            response->set_votestate( Expire ); // 说明你的竞选是过期的，该竞选者已经过时了
            return;
        }

        // 不可能就根据你的任期大于我我就给你投票呀，还需要有后序的检查
        if (request->term() > m_currentTerm)
        {
            m_role = Follower;
            m_currentTerm = request->term();
            m_vectorFor = -1;
        }

        myAssert( request->term() == m_currentTerm, format( "[func--rf{%d}] 前面校验过args.Term==rf.currentTerm，这里却不等", m_id ) );

        // 投票的条件一定是RequestVoteArgs里面的lastLogIndex和lastlogTerm大于或等于投票人
        // 我们需要获得我们最新的term和index
        // 如果比较出来的是失败的，则说明申请投票者的term可能小于投票者的term 或者申请投票者的term =
        // 投票者的term并且申请投票者的logindex小于投票者的logindex
        if (!compareLogIndexAndTerm( request->lastlogindex(), request->lastlogterm() ))
        {
            response->set_term( m_currentTerm );
            response->set_votestate( Voted ); // 我投过票了，不是给你的
            response->set_votegranted( false );
            return;
        }

        // 当我已经投过票了，但是还是收到了其它人的投票请求讷？，一个人只能投票一次
        // 哪些字段标识了我已经投过票了呢？m_vectorFor，如果如果我的m_vectorFor不能与-1了，并且还不等于request的CandidateId
        // m_vectorFor == -1则说明本任期内我还没有投过票，如果不等于-1则说明我已经投过票了
        // 为什么还需要有m_vectorFor
        // !=request->candidateid()可能存在的情况是，我已经给你投票了，但是因为网络原因，你没有收到，你重新发送了一次投票请求
        // 然后你重新发了投票申请，接收到之后我会检查m_vectorFor 如果不等于-1说明我投过票，如果我的m_vectorFor等于你然后我又会重新发送一次
        // 如果不等于就说明上次的投票不是投给你的
        if (m_vectorFor != -1 && m_vectorFor != request->candidateid())
        {
            response->set_term( m_currentTerm );
            response->set_votestate( Voted );
            response->set_votegranted( false );
            return;
        }

        if (m_vectorFor == -1 || m_vectorFor == request->candidateid())
        {
            m_vectorFor = request->candidateid();
            m_lastResetElectionTime = now();
            response->set_term( m_currentTerm );
            response->set_votestate( Voted );
            response->set_votegranted( true );
        }
    }

    // 这就是RPC远程调用的服务的方法；当服务方接收到调用时，就会执行这个调用的方法；
    void Raft::InstallSnapshot( google::protobuf::RpcController *controller,
                                const raftRpcProtoc::InstallSnapshotRequest *request,
                                raftRpcProtoc::InstallSnapshotResponse *response,
                                google::protobuf::Closure *done )
    {
        InstallSnapshotImp( request, response );
        done->Run(); // 这是回调函数的执行
    }

    /**
     * @brief 这个函数是处理leader发送给follower的InstallSnapshotRequest的处理逻辑
     *
     * @param request 这是leader发送过来的快照信息
     * @param response 这是follower对快照信息处理后发回leader的响应数据
     */
    void Raft::InstallSnapshotImp( const raftRpcProtoc::InstallSnapshotRequest *request, raftRpcProtoc::InstallSnapshotResponse *response )
    {
        // 先会检查leader的任期
        std::lock_guard< std::mutex > lock( mutex_ );
        if (request->term() < m_currentTerm)
        {
            response->set_term( m_currentTerm );
            return;
        }

        if (request->term() > m_currentTerm)
        {
            m_currentTerm = request->term();
            m_vectorFor = -1;
            m_role = Follower;
            persist();
        }

        m_role = Follower;
        // 只要收到了leader发送的信息，并且任期是大于等本节点当前的任期，就需要重置选举定时器
        m_lastResetElectionTime = now();

        // 检查快照里面的索引
        if (request->lastsnapshotincludeindex() <= m_lastSnapShotIndex)
        {
            std::cout << "Leader{" << request->leaderid() << "}发送过来的快照，节点{" << m_id << "}已经存在" << std::endl;
            return;
        }

        // 获取本节点最新的日志索引
        auto lastLogIndex = getLastLogIndex();

        // 如果我本地的最新的日志索引是大于快照的最后一个日志索引的，我们会把这个快照给上传的状态机中，因此快照中包含的旧日志在本地中就需要删除来节省空间，因为我们已经提交了这部分的日志的快照了
        if (lastLogIndex > request->lastsnapshotincludeindex())
        {
            // 我们需要删除掉快照中存在的日志，即旧日志
            m_logs.erase( m_logs.begin(), m_logs.begin() + getSlicesIndexFromLogIndex( request->lastsnapshotincludeindex() ) + 1 );
        }
        else
        {
            // 等于的话呢
            m_logs.clear();
        }

        // 更新提交的日志索引
        m_commitIndex = std::max( m_commitIndex, request->lastsnapshotincludeindex() );
        m_lastApplied = std::max( m_lastApplied, request->lastsnapshotincludeindex() );
        m_lastSnapShotIndex = request->lastsnapshotincludeindex();
        m_lastSnapShotTerm = request->lastsnapshotincludeterm();

        // 设置回复
        response->set_term( m_currentTerm );

        // 通过和上层kv数据库通信的结构体将数据上传
        ApplyMsg amsg;
        amsg.SnapshotValid = true;
        amsg.SnapshotIndex = request->lastsnapshotincludeindex();
        amsg.SnapshotTerm = request->lastsnapshotincludeterm();
        amsg.Snapshot = request->data();

        // 使用一个线程来上传数据
        std::thread t( &Raft::pushMsgToKvServer, this, amsg );
        t.detach();
        m_persister->BothSave( persistData(), amsg.Snapshot );
    }

    // 这也是rpc调用的请求方
    void Raft::doElection()
    {

        // 首先判断我自己的身份是不是leader
        if (m_role == Leader) { return; }

        m_role = Candidate;

        // 然后就是任期加1
        m_currentTerm += 1;

        // 给自己投票
        m_vectorFor = m_id;

        persist();

        // 记录我自己的获得的票数
        std::shared_ptr< int > votedNum = std::make_shared< int >( 1 );

        // 重新设置定期是
        m_lastResetElectionTime = now();

        // 然后就是和其它的raft节点进行通信
        for (int i = 0; i < m_peers.size(); ++i)
        {
            if (i == m_id) { continue; }

            // 然后就是需要填充RequestVoteArgs参数了
            // 先获取到我自己最新的logIndex和term
            int lastLogIndex = -1;
            int lastLogTerm = -1;
            getLastLogIndexAndTerm( &lastLogIndex, &lastLogTerm );

            // 然后就是生成RequestVoteArgs对象
            std::shared_ptr< raftRpcProtoc::RequestVoteArgs > requestVoteArgs = std::make_shared< raftRpcProtoc::RequestVoteArgs >();
            requestVoteArgs->set_candidateid( m_id );
            requestVoteArgs->set_term( m_currentTerm );
            requestVoteArgs->set_lastlogindex( lastLogIndex );
            requestVoteArgs->set_lastlogterm( lastLogTerm );

            // 然后就是生成RequestVoteReply对象

            std::shared_ptr< raftRpcProtoc::RequestVoteReply > requestVoteReply = std::make_shared< raftRpcProtoc::RequestVoteReply >();

            // 然后就是要发送请求了，使用raftStub来发送
            std::thread t( &Raft::sendRequestVote, this, i, requestVoteArgs, requestVoteReply, votedNum );

            // 创建了一个线程，因此需要让主线程和子线程分离，不能等着
            t.detach();
        }
    }

    // 这是请求方发送的RPC调用请求
    bool Raft::sendRequestVote( int peer,
                                std::shared_ptr< raftRpcProtoc::RequestVoteArgs > request,
                                std::shared_ptr< raftRpcProtoc::RequestVoteReply > reply,
                                std::shared_ptr< int > votedNum )
    {
        auto start = std::chrono::high_resolution_clock::now();
        std::cout << "[func-sendRequestVote rf{" << m_id << "} 向peer{" << peer << "} 发送RequestVote开始" << std::endl;
        // 从m_peers里面来发送请求
        bool sendRet = m_peers[ peer ]->RequestVote( request.get(), reply.get() );
        auto end = std::chrono::high_resolution_clock::now();

        // std::chrono::duration<double> 用于表示时间的持续时间。
        std::chrono::duration< double > elapsed = end - start;
        std::cout << "[func-sendRequestVote rf{" << m_id << "} 向peer{" << peer << "} 发送RequestVote完毕，耗时{" << elapsed.count()
            << "} ms" << std::endl;

        LOG( INFO ) << "[func-sendRequestVote rf{" << m_id << "} 向peer{" << peer << "} 发送RequestVote完毕，耗时{" << elapsed.count()
            << "} ms";
        if (!sendRet)
        {
            std::cout << "发送失败" << std::endl;
            return false;
        }

        // 对回应进行处理，为什么发送之后就能对回应进行处理讷？这是因为我们的stub继承了RPCChannel类，并且重写了callmethod方法，这个方法在发送完数据之后，开始接收响应数据
        // 并将数据填入到response中，因此我们发送完毕之后，它是用的阻塞读取函数，因此如果没有接收到数据会阻塞，当执行到这里的时候，说明接收完毕了；
        std::lock_guard< std::mutex > lock( mutex_ );

        // 然后就是判断对端的填充的任期
        if (reply->term() > m_currentTerm)
        {
            // 说明对端的任期大于我的，就是会选举失败了
            m_role = Follower;
            m_currentTerm = reply->term();
            // 我不投票了
            m_vectorFor = -1;
        }
        else if (reply->term() <= m_currentTerm)
        {

            // 在raft中，appendEntries和requestVote通信的双方必须保持任期的一致，否则就不会处理
            return true;
        }

        myAssert( reply->term() == m_currentTerm, format( "assert {reply.Term==rf.currentTerm} fail" ) );

        // 然后就是处理里面的数据
        if (!reply->votegranted()) { return true; }

        *votedNum += 1;

        // raft中请求的投票数必须要大于一半才能成为新的领导者
        if (*votedNum >= ( m_peers.size() / 2 ) + 1)
        {
            // 那么我就会成为新的leader(当之前不是leader时)
            // 重新得票数
            *votedNum = 0;

            if (m_role == Leader)
            {
                std::cout << "perr{" << m_id << "} 又一次当选为Leader，当前任期为：" << m_currentTerm << std::endl;
                LOG( INFO ) << "perr{" << m_id << "} 又一次当选为Leader，当前任期为：" << m_currentTerm << std::endl;
            }
            else
            {
                m_role = Leader;

                std::cout << "perr{" << m_id << "} 当选为Leader，当前任期为：" << m_currentTerm << std::endl;
                LOG( INFO ) << "perr{" << m_id << "} 当选为Leader，当前任期为：" << m_currentTerm << std::endl;
            }

            // 每次当选为Leader后都需要发送心跳报文，即AppendEntriesArgs消息，但是里面是没有日志条目的
            // 并且还需要维护nextIndex的值，论文中描述的是nextIndex的值是getLastLogIndex() + 1
            int lastLogIndex = getLastLogIndex();
            for (int i = 0; i < m_nextIndex.size(); ++i)
            {
                m_nextIndex[ i ] = lastLogIndex + 1;
                m_matchIndex[ i ] = 0;
            }

            // 然后就需要创建一个线程来执行心跳报文的发送
            std::thread t( &Raft::doHeartBeat, this );
            t.detach();

            persist();
        }
        return true;
    }

    /**
     * @brief 这是发送心跳报文的函数
     *  心跳报文的内容：
     *  term：当前领导者的任期。
     *  leaderId：领导者的 ID。
     *  prevLogIndex：领导者上一个日志条目的索引，用于跟随者判断日志同步的一致性。
     *  prevLogTerm：领导者上一个日志条目的任期。
     *  request： 日志数据
     *  leaderCommit：领导者已提交的最大日志索引。
     *
     */
    void Raft::doHeartBeat()
    {
        std::lock_guard< std::mutex > lock( m_mutex );

        if (m_role == Leader)
        {
            std::cerr << "leader开始发送心跳报文" << std::endl;
            auto appendNums = std::make_shared< int >( 1 );

            // 发送心跳报文
            for (int i = 0; i < m_peers.size(); ++i)
            {
                if (i == m_id)
                    continue; //这是跳过给自己的发送

                // 每次发送心跳报文时，都会检查每次发送心跳（即 AppendEntries 请求）时，领导者需要检查跟随者的 nextIndex
                // 是否小于领导者的快照索引。如果 nextIndex
                // 小于快照索引，说明该跟随者需要的日志条目已经被领导者删除，并且被快照覆盖。在这种情况下，领导者应该发送快照而不是日志条目，以帮助跟随者同步状态。
                // 通过循环来判断follower的nextIndex是否小于我的最后一个快照的index
                if (m_nextIndex[ i ] <= m_lastSnapShotIndex)
                {
                    std::thread t( &Raft::leaderSendSnapShot, this, i ); // 这也是通过AppendEntries来发送的
                    t.detach();
                    continue;
                }

                auto request = std::make_shared< raftRpcProtoc::AppendEntriesArgs >();
                // 获取prelogIndex和prelogTerm，正常来说应该是我的最后的一个index和term
                int prelogIndex = -1;
                int prelogTerm = -1;
                //这是获取前一个发送的日志的索引和任期，这是通过获取m_nextIndex
                getPreLogIndexAndTerm( i, &prelogIndex, &prelogTerm );

                request->set_term( m_currentTerm );
                request->set_prelogindex( prelogIndex );
                request->set_prelogterm( prelogTerm );
                request->clear_entries();
                request->set_leadercommit( m_commitIndex );
                // 心跳报文中也携带日志条目信息，用来更新日志条目
                // 需要判断前一个日志是在快照中还是在本地的日志条目中;
                // 如果prelogIndex不能与最后一个快照的索引，则说明是在本地的日志中
                if (prelogIndex != m_lastSnapShotIndex)
                {
                    // 从本地中查找prelogIndex下一个index在本地条目的哪
                    for (int j = getSlicesIndexFromLogIndex( prelogIndex ) + 1; j < m_logs.size(); ++j)
                    {
                        // 获取logIndex
                        // 添加一个新的日志条目
                        auto newEntry = request->add_entries();
                        *newEntry = m_logs[ j ];
                    }
                }
                else
                {
                    // 如果相等，则直接把本地的全部加入进去
                    for (auto &item : m_logs)
                    {
                        auto newEntry = request->add_entries();
                        *newEntry = item;
                    }
                }

                int lastLogIndex = getLastLogIndex();

                // 需要将新增的条目全部加入到Entry里面去，因此需要判断preLogindex + 新增的log数是否等于我自己的lastLogIndex数
                myAssert( request->prelogindex() + request->entries_size() == lastLogIndex,
                          format( "appendEntriesArgs.PrevLogIndex{%d}+len(appendEntriesArgs.Entries){%d} != lastLogIndex{%d}",
                                  request->prelogindex(),
                                  request->entries_size(),
                                  lastLogIndex ) );

                // 构造返回值
                const std::shared_ptr< raftRpcProtoc::AppendEntriesReply > appendEntriesReply =
                    std::make_shared< raftRpcProtoc::AppendEntriesReply >();
                appendEntriesReply->set_appstate( Disconnected );

                std::thread t( &Raft::sendAppendEntries, this, i, request, appendEntriesReply, appendNums );
                t.detach();
            }
            m_lastHeartBeatTime = now();
        }
    }

    bool Raft::sendAppendEntries( int peer,
                                  std::shared_ptr< raftRpcProtoc::AppendEntriesArgs > request,
                                  std::shared_ptr< raftRpcProtoc::AppendEntriesReply > reply,
                                  int &appendNum )
    {

        auto start = std::chrono::high_resolution_clock::now();
        std::cout << "[func-sendAppendEntries rf{" << m_id << "} 向peer{" << peer << "} 发送AppendEntries开始" << std::endl;
        // 从m_peers里面来发送请求
        bool sendRet = m_peers[ peer ]->AppendEntries( request.get(), reply.get() );
        auto end = std::chrono::high_resolution_clock::now();

        // std::chrono::duration<double> 用于表示时间的持续时间。
        std::chrono::duration< double > elapsed = end - start;
        std::cout << "[func-sendAppendEntries rf{" << m_id << "} 向peer{" << peer << "} 发送AppendEntries完毕，耗时{" << elapsed.count()
            << "} ms" << std::endl;

        LOG( INFO ) << "[func-sendAppendEntries rf{" << m_id << "} 向peer{" << peer << "} 发送AppendEntries完毕，耗时{" << elapsed.count()
            << "} ms";
        if (!sendRet)
        {
            std::cout << "发送失败" << std::endl;
            return sendRet;
        }

        // 然后就是处理响应了
        std::lock_guard< std::mutex > lock( mutex_ );

        if (reply->term() > m_currentTerm)
        {
            // 退化成follower
            m_role = Follower;
            m_currentTerm = reply->term();
            m_vectorFor = -1;
            return sendRet;
        }
        else if (reply->term() < m_currentTerm)
        {
            DPrintf( "[func -sendAppendEntries  rf{%d}]  节点：{%d}的term{%d}<rf{%d}的term{%d}\n",
                     m_id,
                     peer,
                     reply->term(),
                     m_id,
                     m_currentTerm );
            return sendRet;
        }

        // 如果我不是leader就不要处理响应的数据了
        if (m_role != Leader) { return sendRet; }
        // 判断它是否成功复制了
        if (!reply->success())
        {
            if (reply->updatenextindex() != -100)
            {
                // 然后就需要判断这个updatenextindex,不成功且不等于-100说明之前是匹配的，但是这一次是不匹配的
                // 不是初始值，就需要更新nextindex
                std::cerr << "[func-sendAppendEntries  rf{" << m_id << "}]  返回的日志term相等，但是不匹配，回缩nextIndex[" << peer
                    << "]：{" << reply->updatenextindex() << "}" << std::endl;
                LOG( ERROR ) << "[func-sendAppendEntries  rf{" << m_id << "}]  返回的日志term相等，但是不匹配，回缩nextIndex[" << peer
                    << "]：{" << reply->updatenextindex() << "}";

                m_nextIndex[ peer ] = reply->updatenextindex();
            }
        }
        else
        {
            // 这里是成功的逻辑，如果成功了
            appendNum += 1;
            std::cout << "---------------------------tmp------------------------- 节点{" << peer << "}返回true,当前appendNums为{"
                << appendNum << "个}" << std::endl;
            LOG( INFO ) << "---------------------------tmp------------------------- 节点{" << peer << "}返回true,当前appendNums为{"
                << appendNum << "个}";
            m_matchIndex[ peer ] = std::max( m_matchIndex[ peer ], request->prelogindex() + request->entries_size() );

            // 更新nextIndex数组
            m_nextIndex[ peer ] = m_matchIndex[ peer ] + 1;

            // 日志提交需要超过半数的节点
            if (appendNum >= m_peers.size() / 2 + 1)
            {
                appendNum = 0;
                // 更新提交状态
                if (request->entries_size() > 0 && request->entries( request->entries_size() - 1 ).logterm() == m_currentTerm)
                {
                    std::cout << "---------------------------tmp------------------------- term有log成功提交，更新leader的m_commitIndex "
                        "from{"
                        << m_commitIndex << "} to{" << request->prelogindex() + request->entries_size() << "}" << std::endl;
                    LOG( INFO ) << "---------------------------tmp------------------------- term有log成功提交，更新leader的m_commitIndex "
                        "from{"
                        << m_commitIndex << "} to{" << request->prelogindex() + request->entries_size() << "}";
                    m_commitIndex = std::max( m_commitIndex, request->prelogindex() + request->entries_size() );
                }
            }
        }
        return sendRet;
    }

    void Raft::getLastLogIndexAndTerm( int *lastLogIndex, int *lastLogTerm )
    {

        int lastLogIndexTmp = -1;
        int lastLogTermTmp = -1;
        if (m_logs.empty())
        {
            lastLogIndexTmp = m_lastSnapShotIndex;
            lastLogTermTmp = m_lastSnapShotTerm;
        }
        else
        {
            lastLogIndexTmp = m_logs[ m_logs.size() - 1 ].logindex();
            lastLogTermTmp = m_logs[ m_logs.size() - 1 ].logterm();
        }

        *lastLogIndex = lastLogIndexTmp;
        *lastLogTerm = lastLogTermTmp;
    }

    /**
     * @brief 这是用来比较申请投票者的logindex和logterm是否比投票者的新
     *
     * @param index 这是申请投票者的logindex
     * @param term  这是申请投票者的term
     * @return 这是返回申请投票者的term是否大于投票者的term或者申请投票者的term等于投票者的term且申请投票者的logindex大于投票者的logindex；
     */
    bool Raft::compareLogIndexAndTerm( int index, int term )
    {

        // 需要获取自己的最新的index和term
        int myLogIndex = -1;
        int myLogTerm = -1;
        getLastLogIndexAndTerm( &myLogIndex, &myLogTerm );
        // 需要你的
        if (term > myLogTerm)
            return true;
        else if (term == myLogTerm && index >= myLogIndex) { return true; }
        else { return false; }
    }

    /**
     * @brief 这个函数的主要作用是，根据nextIndex数组来填充preLogIndex和preLogTerm，nextIndex
     * 存储的是领导者需要发送给每个跟随者的下一个日志条目的索引。
     *
     * @param index
     * @param term
     */
    void Raft::getPreLogIndexAndTerm( int peer, int *index, int *term )
    {
        // 日志存在可能是本地的日志或者是日志快照，因此查找索引是可以先判断一个索引是否是日志快照里面的索引讷
        if (m_nextIndex[ peer ] == m_lastSnapShotIndex + 1)
        {
            // 就说明上一次发送的日志条目的最有一个索引就是快照的最后一个索引
            *index = m_lastSnapShotIndex;
            *term = m_lastSnapShotTerm;
        }

        // 如果不是的话，就等于nextIndex-1
        int nextIndex = m_nextIndex[ peer ];
        *index = nextIndex - 1;
        *term = getLogTermFromLogIndex( *index );
    }

    void Raft::applierTicker()
    {

        // 定时的任务,将提交的日志写入到状态机中
        while (true)
        {
            std::unique_lock< std::mutex > lock( mutex_ );
            auto applyMsgs = getApplyLogs();
            lock.unlock();

            if (!applyMsgs.empty()) { for (auto &it : applyMsgs) { applyChan->Push( it ); } }

            // 进行睡眠
            sleepNMilliseconds( ApplyInterval );
        }
    }

    /**
     * @brief 这个函数的作用是将当前当前需要提交到上层应用机的日志，填入到容器中
     *
     * @return std::vector<ApplyMsg>
     */
    std::vector< ApplyMsg > Raft::getApplyLogs()
    {
        std::vector< ApplyMsg > appMsgs;
        while (m_lastApplied < m_commitIndex)
        {
            m_lastApplied++;

            // 需要判断每个提交的日志里面的logindex必须要和m_lastApplied一致，比如当前的m_lastApplied是11，那么m_logs里面11的这条日志里面的logIndex也必须是11；
            // m_lastApplied最后一个条目指向11时，其必须要提交logIndex为11的这个索引，而不是其它的索引，因此需要进行断言检查，看看该日志的索引是否时11，如果不是则说明出错，不会上传
            myAssert( m_logs[ getSlicesIndexFromLogIndex( m_lastApplied ) ].logindex() == m_lastApplied,
                      format( "rf.logs[rf.getSlicesIndexFromLogIndex(rf.lastApplied)].LogIndex{%d} != rf.lastApplied{%d} ",
                              m_logs[ getSlicesIndexFromLogIndex( m_lastApplied ) ].logindex(),
                              m_lastApplied ) );

            ApplyMsg aMsg;
            aMsg.commandValid = true;
            aMsg.SnapshotValid = false;
            aMsg.command = m_logs[ getSlicesIndexFromLogIndex( m_lastApplied ) ].command();
            aMsg.commandIndex = m_lastApplied;
            appMsgs.emplace_back( aMsg );
        }

        return appMsgs;
    }

    void Raft::getState( int *term, bool *isLeader )
    {
        std::lock_guard< std::mutex > lock( m_mutex );
        *term = m_currentTerm;
        *isLeader = ( m_role == Leader );
    }

    void Raft::leaderHearBeatTicker()
    {
        while (true)
        {
            while (m_role != Leader) { usleep( 1000 * HeartBeatTimeOut ); }

            // 原子变量不可中断
            static std::atomic< int32_t > atomicCount = 0;

            // 选择一个合适的睡眠时间
            std::chrono::duration< signed long int, std::ratio< 1, 1000000000 > > suitableSleepTime{};
            std::chrono::system_clock::time_point wakeTime{};
            {
                std::unique_lock< std::mutex > lock( mutex_ );
                wakeTime = now();
                suitableSleepTime = getRandomizedElectionTimeout() + m_lastHeartBeatTime - wakeTime;
                lock.unlock();
            }

            if (std::chrono::duration< double, std::milli >( suitableSleepTime ).count() > 1)
            {
                std::cout << atomicCount << "\033[1;35m leaderHearBeatTicker();函数设置睡眠时间为: "
                    << std::chrono::duration_cast< std::chrono::milliseconds >( suitableSleepTime ).count() << " 毫秒\033[0m"
                    << std::endl;
                // 获取当前时间点
                auto start = std::chrono::steady_clock::now();

                usleep( std::chrono::duration_cast< std::chrono::microseconds >( suitableSleepTime ).count() );
                // std::this_thread::sleep_for(suitableSleepTime);

                // 获取函数运行结束后的时间点
                auto end = std::chrono::steady_clock::now();

                // 计算时间差并输出结果（单位为毫秒）
                std::chrono::duration< double, std::milli > duration = end - start;

                // 使用ANSI控制序列将输出颜色修改为紫色
                std::cout << atomicCount << "\033[1;35m leaderHearBeatTicker();函数实际睡眠时间为: " << duration.count() << " 毫秒\033[0m"
                    << std::endl;
                ++atomicCount;
            }

            if (std::chrono::duration< double, std::milli >( m_lastHeartBeatTime - wakeTime ).count() > 0)
            {
                // 睡眠的这段时间有重置定时器，没有超时，再次睡眠
                continue;
            }
            // DPrintf("[func-Raft::doHeartBeat()-Leader: {%d}] Leader的心跳定时器触发了\n", m_me);
            doHeartBeat();
        }
    }

    void Raft::electionTimeOutTicker()
    {
        while (true)
        {
            // 定期发送心跳报文
            while (m_role == Leader)
            {
                // 睡眠一个心跳超时时间
                usleep( HeartBeatTimeOut );
            }
            // 这是因为选举超时时间一般比心跳超时时间长
            std::chrono::duration< signed long int, std::ratio< 1, 1000000000 > > suitableSleepTime{};
            std::chrono::system_clock::time_point wakeTime{};
            {
                m_mutex.lock();
                wakeTime = now();
                suitableSleepTime = getRandomizedElectionTimeout() + m_lastResetElectionTime - wakeTime;
                m_mutex.unlock();
            }

            if (std::chrono::duration< double, std::milli >( suitableSleepTime ).count() > 1)
            {
                // 获取当前时间点
                auto start = std::chrono::steady_clock::now();

                usleep( std::chrono::duration_cast< std::chrono::microseconds >( suitableSleepTime ).count() );
                // std::this_thread::sleep_for(suitableSleepTime);

                // 获取函数运行结束后的时间点
                auto end = std::chrono::steady_clock::now();

                // 计算时间差并输出结果（单位为毫秒）
                std::chrono::duration< double, std::milli > duration = end - start;

                // 使用ANSI控制序列将输出颜色修改为紫色
                std::cout << "\033[1;35m electionTimeOutTicker();函数设置睡眠时间为: "
                    << std::chrono::duration_cast< std::chrono::milliseconds >( suitableSleepTime ).count() << " 毫秒\033[0m"
                    << std::endl;
                std::cout << "\033[1;35m electionTimeOutTicker();函数实际睡眠时间为: " << duration.count() << " 毫秒\033[0m" << std::endl;
            }

            if (std::chrono::duration< double, std::milli >( m_lastResetElectionTime - wakeTime ).count() > 0)
            {
                // 说明睡眠的这段时间有重置定时器，那么就没有超时，再次睡眠
                continue;
            }
            doElection();
        }
    }

    int Raft::getNewCommandIndex()
    {
        auto lastLogIndex = getLastLogIndex();
        return lastLogIndex + 1;
    }

    void Raft::pushMsgToKvServer( const ApplyMsg &msg ) const { applyChan->Push( msg ); }

    /**
     *  这个函数的作用是leader将快照信息发送给对等体，领导者需要从持久化中读取到快照的信息，然后填入到request中，然后通过RPC调用来发送数据
     *  并或者返回的数据，同时leader需要检查返回的数据，如果正常，则更新matchindex和nextindex数组，否则则需要转变角色
     * @param peer
     */
    void Raft::leaderSendSnapShot( int peer )
    {
        std::unique_lock< std::mutex > lock( m_mutex );
        lock.lock();
        raftRpcProtoc::InstallSnapshotRequest args;
        args.set_leaderid( m_id );
        args.set_term( m_currentTerm );
        args.set_lastsnapshotincludeterm( m_lastSnapShotTerm );
        args.set_lastsnapshotincludeindex( m_lastSnapShotIndex );
        args.set_data( m_persister->ReadSnapShot() );
        raftRpcProtoc::InstallSnapshotResponse reply;
        lock.unlock();
        bool ret = m_peers[ peer ]->InstallSnapshot( &args, &reply );
        std::lock_guard< std::mutex > lock2( m_mutex );
        if (!ret)
            return;

        if (m_role != Leader || m_currentTerm != args.term())
            return;
        //判断reply的term，如果reply的term比自己大了就需要身份变更了
        if (reply.term() > m_currentTerm)
        {
            m_currentTerm = reply.term();
            m_vectorFor = -1;
            m_role = Follower;
            persist();
            m_lastResetElectionTime = now();
            return;
        }
        //因为我们将快照发送给peer了，因此peer的匹配的index就需要变化了,并且下一个需要发送的index就变成了快照的下一个了
        m_matchIndex[ peer ] = args.lastsnapshotincludeindex();
        m_nextIndex[ peer ] = m_matchIndex[ peer ] + 1; //nextIndex ，永远是匹配的index的下一个index

    }

    /**
     * @brief 这个函数的作用是用来完成日志的提交的，即提交到上层状态机中。节点通过判断当前的index是否在其它节点中了，这是通过m_matchIndex数组
     * 来判断的，当该日志在一半以上的节点中存在时，则说明这个日志已被复制到大多数节点中。但是这个节点能被提交到上层状态机的另一个条件是，这个index的任期一定是
     * 当前任期，即只有在当前任期内由领导者追加的日志条目才能被提交。任期不一致则说明出现了领导者变更的情况。
     *  注：提交的一个重要原则：！！！只有在当前任期内由当前领导者追加的日志条目才会被提交。
     */
    void Raft::leaderUpdateCommintIndex()
    {
        m_commitIndex = m_lastSnapShotIndex; //快照每个节点都会有

        //然后从当前logindex开始往前进行统计
        int index = 0;
        for (index = getLastLogIndex(); index > m_lastSnapShotIndex; index--)
        {
            //查每一个节点的match_index,如果满足条件的节点数等于当前raft网络的节点数，则说明这个index是满足情况的
            int sum = 0;
            for (int i = 0; i < m_peers.size(); i++)
            {
                if (i == m_id)
                {
                    sum++;
                    continue;
                }

                if (m_matchIndex[ i ] > index) { sum++; }
            }

            //只有当前任期内由领导者追加的日志条目才会被提交，这是为了维护日志和状态的一致性
            //！！！只有在当前任期内由当前领导者追加的日志条目才会被提交。
            if (sum >= m_peers.size() / 2 + 1 && getLogTermFromLogIndex( index ) == m_currentTerm)
            {
                m_commitIndex = index;
                break;
            }
        }
    }

    /**
     * @brief 这个函数的作用就是检查index对应的本地的log的term，是否等于传递过来的term，用来检查日志是否匹配的
     * @param index 传递过来的需要检查的日志的索引
     * @param term  传递过来的需要比较的任期
     * @return
     */
    bool Raft::mathcLog( int index, int term )
    {
        myAssert( index >= m_lastSnapShotIndex && index <= getLastLogIndex(),
                  format( "不满足：logIndex{%d}>=rf.lastSnapshotIncludeIndex{%d}&&logIndex{%d}<=rf.getLastLogIndex{%d}",
                          index,
                          m_lastSnapShotIndex,
                          index,
                          getLastLogIndex() ) );
        return term == getLogTermFromLogIndex( index );

    }

    /**
     * @brief 这个函数的作用是将节点的数据，例如任期、投票、日志等信息持久化存储起来
     */
    void Raft::persist()
    {
        auto data = persistData();          //这个data就是当前raft节点的信息序列化后的数据，
        m_persister->saveRaftState( data ); //这个命令会将数据保存起来
    }

    /**
     * @brief 这个函数的作用是将节点的当前任期、最后一个快照的index和Term、投票信息以及日志信息保存起来，主要用来保存状态信息，以便后面的节点故障恢复使用
     * @return 返回序列化后的字符串
     */
    std::string Raft::persistData()
    {
        BoostPersistRaftNode raftNode;
        raftNode.m_currentTerm = m_currentTerm;
        raftNode.m_lastSnapShotTerm = m_lastSnapShotTerm;
        raftNode.m_lastSnapshotIndex = m_lastSnapShotIndex;
        raftNode.m_vectorFor = m_vectorFor;
        for (auto &item : m_logs) { raftNode.m_logs.emplace_back( item.SerializeAsString() ); }

        std::stringstream ss;
        boost::archive::text_oarchive oa( ss );
        oa << raftNode;

        return ss.str();
    }

    int Raft::getLogTermFromLogIndex( int logIndex )
    {
        //这个logIndex一定是需要大于等于快照得index，并且也是需要小于等于最新的一个日志的index的
        myAssert( logIndex >= m_lastSnapShotIndex,
                  format( "[func-getSlicesIndexFromLogIndex-rf{%d}]  index{%d} < rf.lastSnapshotIncludeIndex{%d}",
                          m_id,
                          logIndex,
                          m_lastSnapShotIndex ) );

        int lastLogIndex = getLastLogIndex();
        myAssert( logIndex <= lastLogIndex,
                  format( "[func-getSlicesIndexFromLogIndex-rf{%d}]  logIndex{%d} > lastLogIndex{%d}",
                          m_id,
                          logIndex,
                          lastLogIndex ) );
        if (logIndex == m_lastSnapShotIndex)
            return m_lastSnapShotTerm;
        else
            return m_logs[ getSlicesIndexFromLogIndex( logIndex ) ].logterm();
    }

    void Raft::init( std::vector< std::shared_ptr< raftStub > > peers,
                     int id,
                     std::shared_ptr< Persister > persister,
                     std::shared_ptr< LockQueue< ApplyMsg > > applych )
    {
        m_peers = peers;
        m_persister = persister;
        m_id = id;
        this->applyChan = applych;
        m_currentTerm = 0;
        m_role = Follower;
        m_commitIndex = 0;
        m_lastApplied = 0;
        m_lastSnapShotIndex = 0;
        m_vectorFor = -1;
        m_lastSnapShotTerm = 0;
        m_lastHeartBeatTime = now();
        m_lastResetElectionTime = now();
        m_logs.clear();

        for (int i = 0; i < m_peers.size(); ++i)
        {
            m_nextIndex.emplace_back( 0 );
            m_matchIndex.emplace_back( 0 );
        }

        //还可能存在故障恢复的情况，因此我们需要读取之前保存的状态信息
        readPersist( m_persister->ReadRaftState() ); //这里面读取的是序列化后的原始数据
        if (m_lastSnapShotIndex > 0)
        {
            m_lastApplied = m_lastSnapShotIndex; //故障恢复当快照index大于0时
        }

        m_ioManager = std::make_unique<monsoon::IOManager>(FIBER_THREAD_NUM, FIBER_USE_CALLER_THREAD);
        m_ioManager->scheduler([this]() -> void { this->leaderHearBeatTicker(); });
        m_ioManager->scheduler([this]() -> void { this->electionTimeOutTicker(); });

        std::thread t3(&Raft::applierTicker, this);
        t3.detach();

    }

    /**
     * @brief 这个函数的作用就是将客户端发送的命令作为日志条目追加到领导者节点的日志中
     * @param command      这是需要上层状态机(KV-Server)需要执行的命令
     * @param newLogIndex   这是用来存储当前领导者logindex的指针
     * @param newLogTerm    获取领导者的任期，即集群的任期
     * @param isLeader      获取当前raft节点是否是领导者的标识
     */
    void Raft::start( Op command, int *newLogIndex, int *newLogTerm, bool *isLeader )
    {
        std::lock_guard< std::mutex > lock( m_mutex );
        //先判断是不是leader
        if (m_role != Leader)
        {
            *newLogIndex = -1;
            *newLogTerm = -1;
            isLeader = false;
            return;
        }

        //如果是leader就需要，就需要生成新的日志，但是现在不会立即将该日志发送出去，因为日志的发送需要通过心跳消息来进行发送，因此后序的心跳报文会发送该日志
        //生成LogEntry
        raftRpcProtoc::LogEntry newLogEntry;
        newLogEntry.set_command( command );
        newLogEntry.set_logterm( m_currentTerm );
        newLogEntry.set_logindex( getNewCommandIndex() );

        DPrintf( "[func-Start-rf{%d}]  lastLogIndex:%d,command:%s\n", m_id, getLastLogIndex(), &command );
        persist();

        *newLogIndex = newLogEntry.logindex();
        *newLogTerm = newLogEntry.logterm();
        *isLeader = true;

    }

    /**
     * @brief 这个函数是将原始的序列化数据进行反序列化
     * @param data  这是原始的数据
     */
    void Raft::readPersist( std::string data )
    {
        if (data.empty()) { return; }

        std::stringstream iss( data );
        boost::archive::text_iarchive ia( iss );
        BoostPersistRaftNode raftNode;
        ia >> raftNode;

        m_currentTerm = raftNode.m_currentTerm;
        m_vectorFor = raftNode.m_vectorFor;
        m_lastSnapShotIndex = raftNode.m_lastSnapshotIndex;
        m_lastSnapShotTerm = raftNode.m_lastSnapShotTerm;
        m_logs.clear();

        for (auto &item : raftNode.m_logs)
        {
            raftRpcProtoc::LogEntry logEntry;
            logEntry.ParseFromString( item );
            m_logs.emplace_back( logEntry );
        }
    }

    /**
     * @brief 这个函数的作用是更新和保存快照元数据以及截取日志;
     * @param index  这是制作成快照的日志的终点，即该index之后的日志不会制作快照，因此终点一定是在m_lastSnapShotIndex和m_commitIndex之间的
     * @param snapshot  这是制作好的快照数据
     */
    void Raft::makeSnapshot( int index, std::string snapshot )
    {
        //只有没制作过快照的log，并且已经提交了的log才能制作快照
        std::lock_guard< std::mutex > lock( m_mutex );

        if (index <= m_lastSnapShotIndex || index > m_commitIndex)
        {
            DPrintf(
                "[func-Snapshot-rf{%d}] rejects replacing log with snapshotIndex %d as current snapshotIndex %d is larger or "
                "smaller ",
                m_id,
                index,
                m_lastSnapShotIndex );
            return;
        }

        auto lastLogIndex = getLastLogIndex();

        //制造快照
        int newLastSnapShotIndex = index; //当前最新快照的终点
        int newLastSnapShotTerm = getLogTermFromLogIndex( newLastSnapShotIndex );

        //定义容器来存放后序没有进行快照的日志，截断
        std::vector< raftRpcProtoc::LogEntry > logs;
        //因为index是终点，因此index+1 到 lastLogIndex 的日志是没有进行快照的
        for (int i = index + 1; i <= lastLogIndex; ++i)
        {
            raftRpcProtoc::LogEntry log = m_logs[ getSlicesIndexFromLogIndex( i ) ];
            logs.emplace_back( log );
        }

        m_lastSnapShotIndex = newLastSnapShotIndex;
        m_lastSnapShotTerm = newLastSnapShotTerm;
        m_logs = logs;

        m_commitIndex = std::max( m_commitIndex, index );
        m_lastApplied = std::max( m_lastApplied, index );

        //保存快照信息，即持久化保存，保存在文件中
        m_persister->BothSave( persistData(), snapshot);

        DPrintf("[SnapShot]Server %d snapshot snapshot index {%d}, term {%d}, loglen {%d}", m_id, index,
         m_lastSnapShotTerm, m_logs.size());

        //判断，剩余的log的数目 + 最后一个快照的index ?= 之前的最后一个log的索引
        // |--------------日志--------|  ---> |-----快照----|------剩余的日志-------| == lastLogIndex
        myAssert( m_lastSnapShotIndex + m_logs.size() == lastLogIndex, format("len(rf.logs){%d} + rf.lastSnapshotIncludeIndex{%d} != lastLogjInde{%d}", m_logs.size(),
                  m_lastSnapShotIndex, lastLogIndex));
    }
} // namespace raftStroeDataBase